package analysis;

import beans.AdClickEvent;
import beans.AdCountViewByProvince;
import beans.BlackListWarning;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;



/**
 * @author zkq
 * @date 2022/10/6 19:54
 */
//黑名单 报警
public class AdStatisticsByProvince_waring {
    public static void main(String[] args) throws Exception {
        String path = AdCountByProvince.class.getResource("/AdClickLog.csv").getPath();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<AdClickEvent> inputStream = env.readTextFile(path)
                .map(data -> {
                    String[] fields = data.split(",");
                    return new AdClickEvent(new Long(fields[0]), Long.valueOf(fields[1]), fields[2], fields[3], new Long(fields[4]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<AdClickEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<AdClickEvent>() {
                            @Override
                            public long extractTimestamp(AdClickEvent element, long recordTimestamp) {
                                return element.getTimestamp()*1000;
                            }
                        })
                );
        //一个用户点击同一个广告超过一百次 不正常
        SingleOutputStreamOperator<AdClickEvent> fliterStream = inputStream
                .keyBy(new KeySelector<AdClickEvent, Tuple2<Long, Long>>() {
                    @Override
                    public Tuple2<Long, Long> getKey(AdClickEvent value) throws Exception {
                        return Tuple2.of(value.getUserId(), value.getAdId());
                    }
                })
                .process(new FilterBlackListUser(100));
        //按省份统计点击量
        SingleOutputStreamOperator<AdCountViewByProvince> result = fliterStream
                .keyBy(data -> data.getProvince())
                .window(SlidingEventTimeWindows.of(Time.hours(1), Time.seconds(60)))
                .aggregate(new AdCountByProvince.AdCountAgg(), new AdCountByProvince.AdCountResult());
        //输出
        result.print("white");
        fliterStream.getSideOutput(new OutputTag<BlackListWarning>("blacklist"){}).print("black");
        env.execute();
    }
    public static class FilterBlackListUser extends KeyedProcessFunction<Tuple2<Long,Long>,AdClickEvent,AdClickEvent>{
        private Integer clickNum;
        //此状态存一个用户点击一个广告的次数
        private ValueState<Long> clickState;
        //此状态作为一个标记 判断用户是否为黑名单用户
        private ValueState<Boolean> flag;

        //有参构造器
        public FilterBlackListUser(Integer clickNum) {
            this.clickNum = clickNum;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            clickState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("click count", Long.class,0L));
            flag = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("flag", Boolean.class,false));
        }

        @Override
        public void processElement(AdClickEvent value, Context ctx, Collector<AdClickEvent> out) throws Exception {
            Long count = clickState.value();
            //判断为第一次 需要注册定时器 明天0点清空状态重新累计点击次数
            if(count == 0){
                //伦敦时间0点是北京时间8点 算出第二天的0点的时间戳注册定时器
                long time = (ctx.timerService().currentProcessingTime()/(24*60*60*1000)+1)-8*(24*60*60*1000);
                ctx.timerService().registerEventTimeTimer(time);
            }

            OutputTag<BlackListWarning> tag = new OutputTag<BlackListWarning>("blacklist"){};
            //超过一百次报警
            if (count >=100){
                //判断是否已经是黑名单用户 只报警一次 如果已经是 直接过滤就行了不需要再次报警
                if(!flag.value()){
                    flag.update(true);
                    ctx.output(tag,new BlackListWarning(value.getUserId(),value.getAdId(),"点击超过100次"));
                }
                return;
            }
            clickState.update(count + 1);
            out.collect(value);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<AdClickEvent> out) throws Exception {
            //清空状态 重新累计
            clickState.clear();
            flag.clear();
        }
    }
}
